package com.project

import org.apache.spark.sql.{SaveMode, SparkSession}

object MallApplogOds2DwdStep3 {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("MallApplogOds2DwdStep3")
      .config("spark.sql.shuffle.partitions","1")
      .master("local")
      .enableHiveSupport()
      .getOrCreate()

    spark.udf.register("geo",Functions.gps2GeoHashcode)

    /**
     * 存在一种可能性：日志数据对应的表中，在地域维度表中通过geohash码
     * 关联不到区域信息。
     */
    val joined = spark.sql(
      """
        |select
        |  a.*,
        |  b.province,
        |  b.city,
        |  b.region
        |from
        |  tmp.app_log_ods2dwd_step2 a  -- 日志表
        |left join
        |  dim.geohash_area b  -- 地域维表
        |on geo(a.latitude,a.longitude)=b.geohash
        |
        |""".stripMargin)

    joined.createTempView("joined")
    
    // 主输出：就是把关联处理后的日志输出，插入到dwd日志明细表中区
    spark.sql(
      """
        |insert overwrite table dwd.bigdata_mall_app_events partition(dt='2022-07-16')
        |select * from joined
        |
        |""".stripMargin)


    // 从查询结果中挑出关联地域信息失败的gps座标，进行侧输出
    // 以便于后续可以用异步任务去对这些gps座标请求高德来得到地域信息
    spark.sql(
      """
        |select
        |  concat_ws(',',latitude,longitude)
        |from joined
        |where province is null --没有关联上地域信息
        |group by latitude,longitude -- 对经纬度去重
        |
        |""".stripMargin)
      .write.mode(SaveMode.Overwrite).text("hdfs://node01:9000/unknown-gps/2022-07-16")

    spark.close()
  }

}